今日目標

1. 利用 Flask-JWT-Extended 建立 Login API 並測試。
2. 利用 @jwt_required() 建立受保護的 Member Tasks API 並測試。
3. 設定刷新接近過期的 token 流程。

今天就來將昨天學到的內容給實作起來吧！ GOGO!

利用 Flask-JWT-Extended 建立 Login API 並測試

建立輸入輸出規格：

• 同樣利用我們建立好的 Login API 規格 來設定。
  • 因為我們會將 token 設置在 cookie，所以需要建立一個 response。

  # 輸入
  member_login_model = api.model("MembersLogin", {
      "username": fields.String,
      "password": fields.String
  })
  
  # 輸出
  response = jsonify({
    "message": "login successful",
    "id": 0,
    "ok": true
  })
  

建立 Login API：

• Login API 的邏輯流程：
  • 根據 member_login_model 規定輸入格式。
  • 根據輸入的 username 檢查是否存在此 member。
  • 根據輸入的 password 檢查是否正確。
  • 建立 JWT 中的 Private Claims。
  • 建立 access_token。
  • 建立 response 並將 access_token、csrf_access_token 設置到 cookies 中。

    # member_controller.py
     @member_ns.route("/login")
     class MemberLoginAPI(Resource):
    
         @member_ns.expect(member_login_model)  # 根據 member_login_model 規定輸入格式
         def post(self):
    
             member = get_member_by_username(member_ns.payload["username"])
             if not member:
                 return {"message": "User dose not exist"}, 401
             if not check_password_hash(member.password_hash, member_ns.payload["password"] + member.salt):
                 return {"message": "Incorrect password"}, 401
    
             member_data = {"username": member.username, "id": member.id}
             access_token = create_access_token(identity=member, additional_claims=member_data)  # 建立 JWT 中的 Private Claims
             response = jsonify({
               "message": "login successful",
               "id": member.id, 
               "ok": True,
             })
             set_access_cookies(response, access_token) # 將 access_token、csrf_access_token 設置到 cookies 中
             return response
    
     # member_model.py
     def get_member_by_username(username):
       return Member.query.filter_by(username=username).first()
    

• 測試 API：
  • 輸入帳號密碼：
    • 
  • 查看回應、cookies：
    • 

利用 @jwt_required() 建立受保護的 Member Tasks API 並測試

建立輸入輸出規格：

• 利用 API 文件建立輸入輸出。

  • 輸入：只需要網址有會員 id 即可。
  • 輸出：沿用 task_model 的版本但 as_list = True。

  # 輸入
  @member_ns.route("/<int:id>/tasks")
  
  # 輸出
  task_model = api.model("Task", {
  "id": fields.Integer(required=False),
  "member_id": fields.Integer,
  "title": fields.String(required=True),
  "priority": fields.String(required=False),
  "state": fields.String(required=False),
  "start": fields.DateTime(required=False),
  "deadline": fields.DateTime(required=False),
  "description": fields.String(required=False)
  })
  

• Member Tasks API 邏輯流程：

  • 設置 @jwt_required() 要求檢驗 JWT token、csrf token。
  • 利用 task_model 設定輸出格式。
  • 利用網址列中的 member id 驗證會員是否存在。
  • 更新：
    • 從 token 中取得 member id。
    • 與網址的 member id 做核對。
  • 利用 Flask-SQLAlchemy 建立的 relationship 取得該會員的所有 tasks。

    # member_controller.py
    @member_ns.route("/<int:id>/tasks")
    class Protected(Resource):
    
        @jwt_required()
        @member_ns.marshal_with(task_model, as_list=True)
        def get(self, id):
            "Member Tasks"
            member = get_member_by_id(id)
            if not member: abort(400, "Member not found")
            jwt_member_id = get_jwt()["id"]
            if not id == jwt_member_id: abort(403, "Forbidden")
            memberTasks = get_member_tasks(member)
            return memberTasks
    
    # member_model.py
    def get_member_by_id(id):
       return Member.query.filter_by(id=id).first()
    
    def get_member_tasks(member):
       return member.tasks
    
    # models.py
    class Member(db.Model):
      tasks = db.relationship("Task", back_populates="member", cascade="all, delete-orphan")
    class Task(db.Model):
      member = db.relationship("Member", back_populates="tasks")
    

• 測試 API：

  • 先測試沒有 token 狀況
    • 
    • 
  • 測試有 token 狀況
    • 

設定刷新接近過期的 token 流程

根據 Flask-JWT-Extended 文件推薦方式：

• 刷新 token 邏輯流程：
  • 使用 after_request 裝飾器建立每次 response 前先執行刷新流程。
  • 檢測 token 剩餘的有效時間是否小於 30 分鐘：
    • 是：就刷新令牌。
    • 否：回傳原始 response。
  • 若 JWT 本就不符合，回傳原始 response。

  # app > __init__.py
  @app.after_request
  def refresh_expiring_jwts(response):
      try:
          exp_timestamp = get_jwt()["exp"]
          now = datetime.now(timezone.utc)
          target_timestamp = datetime.timestamp(now + timedelta(minutes=30))
          if target_timestamp > exp_timestamp:
              access_token = create_access_token(identity=get_jwt_identity())
              set_access_cookies(response, access_token)
          return response
      except (RuntimeError, KeyError):
          # Case where there is not a valid JWT. Just return the original response
          return response
  

補充

• 因為我們的 Member Tasks API 是使用 GET 方法，所以不會驗證 CSRF Token，因為 CSRF 攻擊一般都是要做些其他動作，如更改資料、新增刪除等。
  • 若是為了取得資料的 CSRF，將 CORS 設定為只給特定的源讀取 Response，非指定源就無法讀取了。

回顧

• 總算將理論跟實作給結合再一起拉~~~!!
• 明天會將 Task 的 API 天加上 JWT 的保護，並且測試 CSRF Token 驗證是否有運作。

更新!

• 感謝 WeHelp 學長 Jesse 提出 Bug！
  • 在 Member Tasks API 中若是登入後可以拿著 token 更換網址 id 後取得別人的 Tasks。
    • 添加驗證 id 的機制：
      • 從 token 中取得 member id。
      • 與網址的 member id 做核對。